package com.lion.hbase.api;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterBase;
import org.apache.hadoop.hbase.regionserver.StoreFile.BloomType;
import org.apache.hadoop.hbase.util.Bytes;

@SuppressWarnings("resource")
public class HbaseAPI {
	private static Configuration conf = null;
	private static HTablePool pool;
	private static final Log LOG = LogFactory.getLog(HbaseAPI.class);
	private final static String CHARSET = "UTF-8";
	static {
		conf = HBaseConfig.getConfiguration();
		pool = new HTablePool(conf, 1024);

	}

	/**
	 * 检查rowkey是否存在
	 * 
	 * @param key
	 * @param tableName
	 * @return
	 */
	public boolean checkExist(byte[] key, String tableName) {
		HTableInterface table = null;
		try {
			table = getHtable(tableName);
			return table.exists(new Get(key));
		} catch (Exception ex) {
			LOG.error(ex, ex);
		} finally {
			try {
				table.close();
			} catch (IOException e) {
				LOG.error(e, e);
			}
		}
		return false;
	}

	/**
	 * 创建表
	 * 
	 * @param tableName
	 * @param familys
	 * @param maxVersion
	 * @return
	 * @throws Exception
	 */
	public static HTableInterface creatTable(String tableName,
			String[] familys, int maxVersion) throws IOException {
		return creatTable(tableName, familys, maxVersion, 0);
	}

	/**
	 * 创建表
	 * 
	 * @param tableName
	 * @param familys
	 * @param maxVersion
	 * @param timetoLive
	 * @return
	 * @throws IOException
	 * @throws Exception
	 */
	public static HTableInterface creatTable(String tableName,
			String[] familys, int maxVersion, int timetoLive)
			throws IOException {
		HBaseAdmin admin = new HBaseAdmin(conf);
		conf.get("fs.default.name");
		if (admin.tableExists(tableName)) {
			LOG.info(String.format("table %s already exists!", tableName));
			return pool.getTable(tableName);
		} else {
			HTableDescriptor tableDesc = new HTableDescriptor(tableName);
			for (int i = 0; i < familys.length; i++) {
				HColumnDescriptor ss = new HColumnDescriptor(familys[i]);
				ss.setBloomFilterType(BloomType.ROW);
				ss.setMaxVersions(maxVersion);
				if (timetoLive > 0) {
					ss.setTimeToLive(timetoLive);
				}
				tableDesc.addFamily(ss);
			}
			admin.createTable(tableDesc);
			return pool.getTable(tableName);
		}
	}

	/**
	 * 删除表
	 * 
	 * @param tableName
	 * @throws Exception
	 */
	public static void deleteTable(String tableName) throws IOException {
		try {
			HBaseAdmin admin = new HBaseAdmin(conf);
			if (admin.tableExists(tableName)) {
				admin.disableTable(tableName);
				admin.deleteTable(tableName);
			}
			HConnectionManager.deleteConnection(conf, false);
		} catch (Exception e) {
			LOG.error(e, e);
		}
	}

	public static HTableInterface getHtable(String tableName)
			throws IOException {
		HTableInterface table = null;
		table = pool.getTable(tableName);
		return table;
	}

	public static List<String> getAllTableName() {
		try {
			HBaseAdmin admin = new HBaseAdmin(conf);
			HTableDescriptor[] tables = admin.listTables();
			List<String> tableNames = new ArrayList<String>();
			for (HTableDescriptor table : tables) {
				tableNames.add(new String(table.getName(), CHARSET));
			}
			return tableNames;
		} catch (Exception e) {
			LOG.error(e,e);
		}
		return new ArrayList<String>();
	}

	public static HTableDescriptor[] listTables(String regex) {
		try {
			HBaseAdmin admin = new HBaseAdmin(conf);
			HTableDescriptor[] tables = null;
			if (regex != null) {
				tables = admin.listTables(regex);
			} else {
				tables = admin.listTables();
			}
			return tables;
		} catch (Exception e) {
			LOG.error(e,e);
		}
		return new HTableDescriptor[0];
	}

	public static HTableDescriptor[] listTables() {
		return listTables(null);
	}

	public static void addRecord(String tableName, byte[] rowKey,
			String family, String qualifier, String value) throws IOException {
		HTableInterface table = getHtable(tableName);
		addRecord(table, rowKey, family, qualifier, value);
	}

	public static void addRecord(HTableInterface table, byte[] rowKey,
			String family, String qualifier, String value) throws IOException {
		try {
			Put put = new Put(rowKey);
			if (value == null){
				value = "";
			}
			put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier),
					Bytes.toBytes(value));
			table.put(put);
		} catch (Exception ex) {
			LOG.error(ex, ex);
		} finally {
			returnTable(table);
		}
	}

	/**
	 * 组装put，提交后需要归还表
	 * 
	 * @param table
	 * @param rowKey
	 * @param family
	 * @param qualifier
	 * @param value
	 * @return
	 */
	public static Put makePut(HTableInterface table, byte[] rowKey,
			String family, String qualifier, String value) {
		Put put = new Put(rowKey);
		if (value == null)
			value = "";
		put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier),
				Bytes.toBytes(value));
		return put;
	}
	
	public static Put makePut(HTableInterface table, byte[] rowKey,
			byte[] family, byte[] qualifier, byte[] value) {
		Put put = new Put(rowKey);
		put.add(family, qualifier,value);
		return put;
	}

	public static void batchRecord(String tableName, List<Put> data) {
		HTableInterface table;
		try {
			table = getHtable(tableName);
			batchRecord(table, data);
		} catch (IOException e) {
			LOG.error(e, e);
		}
	}

	public static void batchRecord(HTableInterface table, List<Put> data) {
		try {
			table.put(data);
		} catch (Exception ex) {
			LOG.error(ex, ex);
		} finally {
			returnTable(table);
		}
	}

	public static void returnTable(HTableInterface table) {
		if (table != null) {
			try {
				table.close();
			} catch (IOException e) {
				LOG.error(e, e);
			}
		}
	}

	/**
	 * 清空表
	 * 
	 * @param tableName
	 */
	public static void truncate(String tableName) {
		try {
			HBaseAdmin admin = new HBaseAdmin(conf);
			if (admin.tableExists(tableName)) {
				HTableDescriptor htd = admin.getTableDescriptor(tableName
						.getBytes());
				admin.disableTable(tableName);
				admin.deleteTable(tableName);
				admin.createTable(htd);
			} else {
				LOG.info("not find table" + tableName);
			}
		} catch (Exception e) {
			LOG.error(e, e);
		}
	}

	public static List<Result> scan(HTableInterface table, String startRow,
			String stopRow, String family, Collection<String> fields,
			List<Filter> filter) {
		ResultScanner resultScan = null;
		List<Result> result = new ArrayList<Result>();
		try {
			Scan scan = new Scan(startRow.getBytes(), stopRow.getBytes());
			scan.setMaxVersions();
			scan.setCaching(100);
			scan.addFamily(family.getBytes(CHARSET));
			if (filter != null) {
				for (Filter f : filter) {
					scan.setFilter(f);
				}
			}
			if (fields != null) {
				for (String f : fields) {
					scan.addColumn(family.getBytes(CHARSET), f.getBytes(CHARSET));
				}
			}
			resultScan = table.getScanner(scan);
			Iterator<Result> dayResult = resultScan.iterator();
			while (dayResult.hasNext()) {
				Result r = dayResult.next();
				result.add(r);
			}
			return result;
		} catch (Exception e) {
			LOG.error(e, e);
		} finally {
			try {
				if (resultScan != null)
					resultScan.close();
				if (table != null)
					table.close();
			} catch (IOException e) {
				LOG.error(e, e);
			}
		}
		return result;
	}

	public static List<Result> scan(HTableInterface table, String startRow,
			String stopRow, String family, Collection<String> fields) {
		return scan(table, startRow, stopRow, family, fields, null);
	}

	public static KeyValue[] getOneRecord(String tableName, byte[] rowKey)
			throws IOException {
		return getOneRecord(getHtable(tableName), rowKey);
	}

	public static KeyValue[] getOneRecord(HTableInterface table, byte[] rowKey)
			throws IOException {
		try {
			Get get = new Get(rowKey);
			Result rs = table.get(get);
			return rs.raw();
		} catch (Exception ex) {
			LOG.error(ex, ex);
			return null;
		} finally {
			returnTable(table);
		}
	}

	public static KeyValue[] getOneRecord(String tableName, Get get)
			throws IOException {
		HTableInterface table = null;
		try {
			table = getHtable(tableName);
			Result rs = table.get(get);
			return rs.raw();
		} catch (Exception ex) {
			LOG.error(ex + ",table name:" + tableName, ex);
			return null;
		} finally {
			returnTable(table);
		}
	}

	public static ResultScanner getAllRecord(String tableName) {
		HTableInterface table = null;
		try {
			table = getHtable(tableName);
			Scan s = new Scan();
			s.setCaching(100);
			s.setMaxVersions(5);
			return table.getScanner(s);
		} catch (Exception e) {
			LOG.error(e, e);
		}
		return null;
	}

	public static ResultScanner findByFilter(String tableName, String family,
			String qualifier, String value, FilterBase filter)
			throws IOException {
		HTableInterface table = null;
		try {
			table = getHtable(tableName);
			Scan s = new Scan();
			s.setCaching(100);
			s.setFilter(filter);
			return table.getScanner(s);
		} catch (Exception ex) {
			LOG.error(ex, ex);
			return null;
		} finally {
			returnTable(table);
		}
	}
}
